package cn.trigram.stream;

import cn.trigram.stream.function.TerConsumer;
import java.util.Collection;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

@Slf4j
public class StreamFuncUtil {

    /**
     * 对于oracle in查询1000条分区查询
     *
     * @param collection
     * @param mapper
     * @return {@link List<R> }
     */
    public static <T, R> Collection<R> partitionOfOracleIn(Collection<T> collection, Function<? super List<T>, ? extends List<R>> mapper) {
        int size = CollectionUtils.size(collection);
        Collection<T> finalList = CollectionUtils.emptyIfNull(collection);
        return Stream.iterate(0, n -> n + 1)
                // 防止数据量过大，in查询超出1000个报错
                .limit(Double.valueOf(Math.ceil((size / 999d))).longValue())
                .map(step -> finalList.stream().skip(step * 999L).limit(999L)
                        .collect(Collectors.toList())
                )
                .map(mapper)
                .flatMap(Collection::stream)
                .collect(Collectors.toList());
    }

    /**
     * 对于oracle in查询1000条分区查询
     * 注意：内部使用并发stream，请注意ThreadLocal或者事务的使用
     *
     * @param collection
     * @param mapper
     * @return {@link List<R> }
     */
    public static <T, R> Collection<R> partitionOfOracleInByParallel(Collection<T> collection, Function<? super List<T>, ? extends List<R>> mapper) {
        int size = CollectionUtils.size(collection);
        Collection<T> finalList = CollectionUtils.emptyIfNull(collection);
        return Stream.iterate(0, n -> n + 1)
                // 防止数据量过大，in查询超出1000个报错
                .limit(Double.valueOf(Math.ceil((size / 999d))).longValue())
                .map(step -> finalList.stream().skip(step * 999L).limit(999L)
                        .collect(Collectors.toList())
                )
                .parallel()
                .map(mapper)
                .flatMap(Collection::stream)
                .collect(Collectors.toList());
    }

    public static <T, R> Function<T, R> consumeExceptionFunction(Function<T, R> mapper, BiConsumer<T, ? super Throwable> errorConsumer) {
        return in -> {
            try {
                return mapper.apply(in);
            } catch (Throwable throwable) {
                errorConsumer.accept(in, throwable);
            }
            return null;
        };
    }

    public static <T, R> Function<T, R> consumeExceptionFunction(Function<T, R> mapper) {
        return consumeExceptionFunction(mapper, (t, throwable) -> log.error("消费异常：", throwable));
    }

    public static <T, U, R> BiFunction<T, U, R> consumeExceptionBiFunction(BiFunction<T, U, R> mapper, TerConsumer<T, U, ? super Throwable> errorConsumer) {
        return (t, u) -> {
            try {
                return mapper.apply(t, u);
            } catch (Throwable throwable) {
                errorConsumer.accept(t, u, throwable);
            }
            return null;
        };
    }

    public static <T, U, R> BiFunction<T, U, R> consumeExceptionBiFunction(BiFunction<T, U, R> mapper) {
        return consumeExceptionBiFunction(mapper, (t, u, throwable) -> log.error("消费异常：", throwable));
    }

    public static <T, R> Consumer<T> consumeExceptionConsumer(Consumer<T> mapper, BiConsumer<T, ? super Throwable> errorConsumer) {
        return t -> {
            try {
                mapper.accept(t);
            } catch (Throwable throwable) {
                errorConsumer.accept(t, throwable);
            }
        };
    }

    public static <T, R> Consumer<T> consumeExceptionConsumer(Consumer<T> mapper) {
        return consumeExceptionConsumer(mapper, (t, throwable) -> log.error("消费异常：", throwable));
    }

    public static <T, U> BiConsumer<T, U> consumeExceptionBiConsumer(BiConsumer<T, U> mapper, TerConsumer<T, U, ? super Throwable> errorConsumer) {
        return (t, u) -> {
            try {
                mapper.accept(t, u);
            } catch (Throwable throwable) {
                errorConsumer.accept(t, u, throwable);
            }
        };
    }

    public static <T, U> BiConsumer<T, U> consumeExceptionBiConsumer(BiConsumer<T, U> mapper) {
        return consumeExceptionBiConsumer(mapper, (t, u, throwable) -> log.error("消费异常：", throwable));
    }


}
